package skewreduce.seaflow;

import skewreduce.framework.EaggDriver;
import skewreduce.framework.logical.LFinalizeOp;
import skewreduce.framework.logical.LMergeOp;
import skewreduce.framework.logical.LPartitionOp;
import skewreduce.framework.logical.LSampleOp;
import skewreduce.framework.logical.LProcessOp;
import skewreduce.framework.physical.PPigOp;
import skewreduce.framework.physical.PhysicalOp;
import skewreduce.framework.physical.UniformSample;
import skewreduce.lib.Cube2;
import skewreduce.lib.Partition;

public class SeaflowFoF extends EaggDriver {
	int dataType;

    public String getAppName() { return "SeaflowFoF"; }

    public LSampleOp createLSampleOp() {
    	if ( getConf().getBoolean("seaflow.sample.uniform",true) )
    		return UniformSample.createLogicalOp(this);
    	
    	return new LSampleOp(this) {
    		@Override
    		protected PhysicalOp createInstance() {
    			return new SeaflowFoFSample();
    		}
		};
    }
    
    public LPartitionOp createLPartitionOp() {
        return new LPartitionOp(this) {
            @Override
            protected PhysicalOp createInstance() {
                return new SeaflowFoFPartition();
            }
        };
    }

    public LProcessOp createLWorkOp() {
        return new LProcessOp(this) {
            @Override
            protected PhysicalOp createInstance() {
                return new LocalFoF();
            }
        };
    }

    public LMergeOp createLMergeOp() {
        return new LMergeOp(this) {
            @Override
            protected PhysicalOp createInstance() {
            	return new SeaflowFoFMerge();
            }
        };
    }

    public LFinalizeOp createLFinalizeOp() {
        return new LFinalizeOp(this) {
            @Override
            protected PhysicalOp createInstance() {
                return new PPigOp(); // FIXME
            }
        };
    }

    @Override
    protected int handleOption(String[] args,int i) {
        if ( "-numPart".equals(args[i]) ) {
            if ( i+1 < args.length ) {
                int n = Integer.parseInt(args[++i]);
                System.err.println("Overriding initial number of partitions to "+n);
                getConf().setInt("skewreduce.partition.initialpartitions",n);
                return 1;
            } else {
                System.err.println("missing -numPart parameter");
            }
        } else if ( "-dataType".equals(args[i]) ) {
        	dataType = Integer.parseInt(args[++i]);
            return 1;
        }
        return -1;
    }

	@Override
	public Partition getRootPartition() {
		//126722,131702
		return ( dataType == 0 ) ? new Cube2("0;0;0,0,0;65536,65536,65536;0") : new Cube2("0;0;126722,0,0;131702,65536,65536;0");
	}
}
